package club.monkeywood.ad.dmp.tag

import club.monkeywood.ad.dmp.util.{JedisPools, TagsUtils}
import org.apache.spark.SparkConf
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}

import scala.collection.mutable.ListBuffer


object Tags4CtxWithMultiUserId extends App {

	if (args.length != 5) {
		println(
			"""
				|cn.dmp.tags.Tags4Ctx
				|参数：
				| 日志输入路径
				| APP字典文件路径
				| 停用词库
				| 日期
				| 输出路径
			""".stripMargin)
		sys.exit()
	}

	val Array(inputPath, dictFilePath, stopWordsFilePath, day, outputPath) = args

	// 2 创建sparkconf->SparkSession
	val sparkConf = new SparkConf()
	sparkConf.setAppName(s"${this.getClass.getSimpleName}")
	sparkConf.setMaster("local[*]")
	//使用KryoSerializer更快
	// RDD 序列化到磁盘 worker与worker之间的数据传输
	sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
	//避免字段名过长报错
	sparkConf.set("spark.debug.maxToStringFields", "100")
	val ss = SparkSession.builder()
		.appName(s"${this.getClass.getSimpleName}")
		.config(sparkConf)
		.getOrCreate()

	// 字典文件---appMapping
	// key:appId
	// value:appName
	val dictMap = ss.sparkContext.textFile(dictFilePath).map(line => {
		val fields = line.split("\t", -1)
		(fields(3), fields(1))
	}).collect().toMap

	// 字典文件 --- stopwords
	val stopWordsMap = ss.sparkContext.textFile(stopWordsFilePath)
												.map((_, 0))  //map比数组性能更好，只用key，不用value,value都取0
												.collect()
												.toMap

	// 将字典数据广播executor
	val broadcastAppDict = ss.sparkContext.broadcast(dictMap)
	val broadcastStopWordsDict = ss.sparkContext.broadcast(stopWordsMap)

	// 读取日志parquet文件
	// parquet文件中有元数据，可以使用sql语句
	val baseData: RDD[Row] = ss.read.parquet(inputPath)
		.where(TagsUtils.hasSomeUserIdConditition)  //过滤掉没有用户id的数据
		.rdd

	/**
		* 1)构建顶点集合：---------------------------------------------------
		*/
  val uv = baseData.mapPartitions(par => {

	  val jedis = JedisPools.getConnection()

	  val userAndTagsOfPar: Iterator[(Long, (String, List[(String, Int)]))] = par.flatMap(row => {

		  // 行数据进行标签化处理
		  // 广告标签
		  val ads = Tags4Ads.makeTags(row)

		  val apps = Tags4App.makeTags(row, broadcastAppDict.value)
		  //设备标签
		  val devices = Tags4Devices.makeTags(row)
		  //关键字标签
		  val keywords = Tags4KeyWords.makeTags(row, broadcastStopWordsDict.value)
		  //用户id标签
		  //返回的来源于多种渠道的用户id
		  val allUserId: List[String] = TagsUtils.getAllUserId(row).toList

		  // 商圈的标签
		  // 从redis商圈知识库中取经纬度对应的商圈标签
		  val business = Tags4Business.makeTags(row, jedis)

		  //用户id暂时未用，故取第一个id占位即可
		  val tagList: List[(String, Int)] = (ads ++ apps ++ devices ++ keywords ++ business).toList

		  val userAndTags: List[(Long, (String, List[(String, Int)]))] = allUserId.map(uid=>{
		    var oneUserAndTags = (allUserId(0).hashCode().toLong, (uid, tagList))
			  if(uid!=allUserId(0)){
				  oneUserAndTags = (uid.hashCode().toLong, (uid, List.empty[(String, Int)]))
			  }
			  oneUserAndTags
		  })

		  userAndTags

	  })

	  jedis.close()
	  userAndTagsOfPar
  })

	/**
		* 2)构建边集合
		*
		*/
	val ue = baseData.flatMap(row=>{
		//用户id标签
		//返回的来源于多种渠道的用户id
		val allUserId: List[String] = TagsUtils.getAllUserId(row).toList

		val edge: List[Edge[Int]]= allUserId.map(uId=>{
			Edge(allUserId(0).hashCode.toLong, uId.hashCode.toLong, 0)
		})

		edge
	})

	/**
		* 3）构建图
		*/
	println("3)构造图--------------------------------------------------")
	val graph = Graph(uv, ue)
	val cc = graph.connectedComponents().vertices

	val mid: RDD[(VertexId, (Set[String], List[(String, Int)]))] = cc.join(uv)
  		.map {
			  case (vertexId, (commonVertexid, (userId, tags))) => (commonVertexid, (Set(userId), tags))
		  }
  	  .reduceByKey{
		    case ((userId1, tags1), (userId2, tags2)) => ( userId1++userId2, tags1++tags2 )
	    }

	val ret: RDD[(Set[String], List[(String, Int)])] = mid.map{
		case (vertexId, (userIds, tags)) => {
				val tagList: List[(String, Int)] = tags.groupBy(_._1)
	        .mapValues(_.foldLeft(0)(_+_._2))
	        .toList
				(userIds, tagList)
			}
		}

	ret.foreach(println)

	//uv.saveAsTextFile(outputPath)

	ss.close()

}
